// Project name:JavaRedis
// Creator:muhong
// Date time:2022/12/8,9:52 下午
// Name:NioEventLoop

package org.gmh.channel.nio;

import org.gmh.channel.EventLoop;
import org.gmh.channel.SelectStrategy;
import org.gmh.channel.SingleThreadEventLoop;
import org.gmh.util.IntSupplier;
import org.gmh.util.concurrent.Future;
import org.gmh.util.concurrent.RejectedExceptionHandler;
import org.gmh.util.concurrent.ScheduledFuture;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.nio.channels.CancelledKeyException;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.util.Iterator;
import java.util.List;
import java.util.Set;
import java.util.concurrent.Callable;
import java.util.concurrent.Executor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;

public final class NioEventLoop extends SingleThreadEventLoop {
    private final static Logger logger = LoggerFactory.getLogger(NioEventLoop.class);

    private Selector selector;

    private final SelectStrategy selectStrategy;

    private final IntSupplier selectNowSupplier = new IntSupplier() {
        @Override
        public int get() throws Exception {
            return selectNow();
        }
    };

    private static final long AWAKE = -1L;
    private static final long NONE = Long.MAX_VALUE;

    private final AtomicLong nextWakeupNanos = new AtomicLong(AWAKE);

    private volatile int ioRatio = 50;
    private int cancelledKeys;
    private boolean needToSelectAgain;

    private static final int MIN_PREMATURE_SELECTOR_RETURNS = 3;

    public NioEventLoop(NioEventLoopGroup parent, Executor executor,
                        SelectStrategy strategy,
                        RejectedExceptionHandler rejectedExceptionHandler) {
        super(parent, executor, false, DEFAULT_MAX_PENDING_TASKS, rejectedExceptionHandler);

        try {
            this.selector = Selector.open();
        } catch (IOException e) {
            e.printStackTrace();
        }
        this.selectStrategy = strategy;
    }

    @Override
    public <V> Future<V> newSucceededFuture(V result) {
        return null;
    }

    @Override
    public <V> Future<V> newFailedFuture(Throwable cause) {
        return null;
    }

    @Override
    public EventLoop next() {
        // No-OP
        return null;
    }

    @Override
    public ScheduledFuture<?> schedule(Runnable command, long delay, TimeUnit unit) {
        return null;
    }

    @Override
    public <V> ScheduledFuture<V> schedule(Callable<V> callable, long delay, TimeUnit unit) {
        return null;
    }

    @Override
    public ScheduledFuture<?> scheduleAtFixedRate(Runnable command, long initialDelay, long period, TimeUnit unit) {
        return null;
    }

    @Override
    public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command, long initialDelay, long delay, TimeUnit unit) {
        return null;
    }

    Selector selector() {
        return this.selector;
    }

    @Override
    protected void run() {
        int selectCnt = 0;
        logger.info("[start] nio event loop working");
        for (; ; ) {
            try {
                int strategy;
                try {
                    strategy = selectStrategy.calculateStrategy(selectNowSupplier, hasTasks());
                    switch (strategy) {
                        case SelectStrategy.CONTINUE:
                            continue;
                        case SelectStrategy.BUSY_WAIT:
                        case SelectStrategy.SELECT:
                            long curDeadlineNanos = NONE;
                            nextWakeupNanos.set(curDeadlineNanos);
                            try {
                                if (!hasTasks()) {
                                    // 这里可能长期阻塞，需要在添加任务那里wakeup
                                    strategy = select(curDeadlineNanos);
                                }
                            } finally {
                                nextWakeupNanos.lazySet(curDeadlineNanos);
                            }
                        default:
                    }
                } catch (IOException e) {
                    // rebuild selector
                    selectCnt = 0;
                    handleLoopException(e);
                    continue;
                }
                selectCnt++;
                cancelledKeys = 0;
                needToSelectAgain = false;
                final int ioRatio = this.ioRatio;
                boolean ranTasks;
                if (ioRatio == 100) {
                    try {
                        if (strategy > 0) { // strategy大小等于key的数量
                            processSelectedKey();
                        }
                    } finally {
                        ranTasks = runAllTask();
                    }
                } else if (strategy > 0) {
                    final long ioStartTime = System.nanoTime();
                    try {
                        processSelectedKey();
                    } finally {
                        final long ioTime = System.nanoTime() - ioStartTime;
                        ranTasks = runAllTask(ioTime * (100 - ioRatio) / ioRatio);
                    }
                } else {
                    ranTasks = runAllTask(0);
                }
                if (ranTasks || strategy > 0) {
                    if (selectCnt > MIN_PREMATURE_SELECTOR_RETURNS) {
                        logger.debug("Selector.select() returned prematurely {} times in a row for Selector {}", selectCnt - 1, selector);
                    }
                    selectCnt = 0;
                }
            } catch (CancelledKeyException e) {
                logger.debug("{} raised by a selector", selector, e);
            } catch (Error e) {
                throw e;
            } catch (Throwable t) {
                handleLoopException(t);
            } finally {
                try {
                    if (isShutdown()) {
                        closeAll();
                    }
                } catch (Error e) {
                    throw e;
                } catch (Throwable t) {
                    handleLoopException(t);
                }
            }
        }
    }

    private void closeAll() {
//        selectAgain();
//        Set<SelectionKey> keys = selector.keys();
//        Collection<AbstractNioChannel> channels = new ArrayList<AbstractNioChannel>(keys.size());
    }

    private void selectAgain() {
        needToSelectAgain = false;
        try {
            selector.selectNow();
        } catch (Throwable t) {
            logger.warn("Failed to update SelectionKeys.", t);
        }
    }

    private static void handleLoopException(Throwable t) {
        logger.warn("Unexpected exception in the selector loop.", t);
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            //
        }
    }

    private int select(long deadlineNanos) throws IOException {
        logger.debug("start to select");
        if (deadlineNanos == NONE) {
            return selector.select();
        }
        long timeoutMills = (deadlineNanos + 995000L) / 1000000L;
        return timeoutMills <= 0 ? selector.selectNow() : selector.select(timeoutMills);
    }

    private void processSelectedKey() {
        Set<SelectionKey> keySet = selector.selectedKeys();
        if (keySet.isEmpty()) {
            return;
        }
        logger.info("process event:" + keySet.size());
        Iterator<SelectionKey> iterator = keySet.iterator();
        while (iterator.hasNext()) {
            SelectionKey key = iterator.next();
            iterator.remove();
            //
            final Object a = key.attachment();
            if (a instanceof AbstractNioChannel) {
                AbstractNioChannel channel = (AbstractNioChannel) key.attachment();
                AbstractNioChannel.NioUnsafe unsafe = channel.unsafe();
                int readyOps = key.readyOps();
                if ((readyOps & (SelectionKey.OP_ACCEPT | SelectionKey.OP_READ)) != 0) {
                    unsafe.read();
                }
            } else {
                logger.info(a.getClass().getName());
            }
        }
    }

    @Override
    protected void wakeup(boolean inEventLoop) {
        if (!inEventLoop && nextWakeupNanos.getAndSet(AWAKE) != AWAKE) {
            selector.wakeup();
        }
    }

    void cancel(SelectionKey key) {
        key.cancel();
    }

    @Override
    public void shutdown() {

    }

    @Override
    public List<Runnable> shutdownNow() {
        return null;
    }

    @Override
    public boolean isShutdown() {
        return false;
    }

    @Override
    public boolean isTerminated() {
        return false;
    }

    @Override
    public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException {
        return false;
    }

    int selectNow() throws IOException {
        return selector.selectNow();
    }
}
